package zjs.dc.controller;

import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;

/**
 * @Class:
 * @Author: gaojian
 * @Description: 针对企查查t_auction表和中数t_judicialauction表进行数据映射开发
 * @Date: 2021/12/20
 */

public class TAuction {
    public static void main(String[] args) {
        SparkSession spark = SparkSession.builder().appName("t_auction").enableHiveSupport().getOrCreate();
        spark.sparkContext().setLogLevel("ERROR");
        Dataset<Row> zs_data = spark.sql("select * from a_dc_ep_ods.zs_t_judicialauction");
        long zs_count = zs_data.count();
        if(zs_count>0){
            // 处理步骤说明 1.UPDATE 2.DELETE
            // 1.UPDATE
            spark.sql("select * from a_dc_ep_ods.zs_t_judicialauction a where length(a.jobid)<10").createOrReplaceTempView("zs_t_auction");

            Dataset<Row> update_data = spark.sql("" +
                    "SELECT \n" +
                    "md5(CONCAT(UPPER(TRIM(B.COMPANY_NAME)),UPPER(TRIM(A.CREDITCODE)),UPPER(TRIM(\n" +
                    "IF(A.AUCTION_ITEM_NAME = '' OR A.AUCTION_ITEM_NAME IS NULL,A.AUCTION_NOTICE,A.AUCTION_ITEM_NAME))),\n" +
                    "DATE_FORMAT(IF(A.PUBLISH_DATE = '' OR A.PUBLISH_DATE IS NULL,'1900-01-01',A.PUBLISH_DATE),'yyyy-MM-dd'))) AS ID,\n" +
                    "B.KEY_NO,\n" +
                    "B.COMPANY_ID,\n" +
                    "B.COMPANY_NAME,\n" +
                    "md5(CONCAT(UPPER(TRIM(B.COMPANY_NAME)),UPPER(TRIM(A.CREDITCODE)),UPPER(TRIM(\n" +
                    "IF(A.AUCTION_ITEM_NAME = '' OR A.AUCTION_ITEM_NAME IS NULL,A.AUCTION_NOTICE,A.AUCTION_ITEM_NAME))),\n" +
                    "DATE_FORMAT(IF(A.PUBLISH_DATE = '' OR A.PUBLISH_DATE IS NULL,'1900-01-01',A.PUBLISH_DATE),'yyyy-MM-dd'))) AS AUCTIONS_ID,\n" +
                    "IF(A.AUCTION_ITEM_NAME = '' OR A.AUCTION_ITEM_NAME IS NULL,A.AUCTION_NOTICE,A.AUCTION_ITEM_NAME) AS NAME,\n" +
                    "A.COURT_NAME AS EXECUTE_GOV,\n" +
                    "A.SALE_PRICE AS YI_WU,\n" +
                    "CONCAT(A.AUCTION_START_TIME,' - ',A.AUCTION_END_TIME) AS ACTION_REMARK,\n" +
                    "MARK_TYPE AS TYPE,\n" +
                    "A.JOBID, '0' AS ISADD \n" +
                    "FROM (\n" +
                    "    select j.* from zs_t_auction j \n" +
                    "    inner join a_dc_ep_ods.zs_t_judicialauction_del del on upper(trim(del.entid)) = upper(trim(j.entid))\n" +
                    "    where del.creditcode is not null and del.creditcode != '' and length(del.jobid)<10\n" +
                    ") A INNER JOIN \n" +
                    "A_DC_EP_ODS.T_ECI_COMPANY B \n" +
                    "ON UPPER(TRIM(B.CREDIT_CODE)) = UPPER(TRIM(A.CREDITCODE)) ");
            update_data.createOrReplaceTempView("tmp_t_auction");
            spark.sql("insert overwrite table a_dc_ep_incr.t_auction select * from tmp_t_auction");



            // 2.DELETE

            Dataset<Row> delete_data = spark.sql("" +
                    "select \n" +
                    "d.id,\n" +
                    "d.key_no,\n" +
                    "d.company_id,\n" +
                    "d.company_name,\n" +
                    "d.auctions_id,\n" +
                    "d.name,\n" +
                    "d.execute_gov,\n" +
                    "d.yi_wu,\n" +
                    "d.action_remark,\n" +
                    "d.type,\n" +
                    "del.jobid,'-1' as isadd\n" +
                    "from (select del.* from a_dc_ep_ods.zs_t_judicialauction_del del \n" +
                    "left join zs_t_auction i on upper(trim(i.entid)) = upper(trim(del.entid))\n" +
                    "where i.entid is null and del.creditcode != '' and del.creditcode is not null and length(del.jobid)<10) del\n" +
                    "inner join (select a.*,c.credit_code from a_dc_ep_ods.t_eci_company c \n" +
                    "inner join a_dc_ep_dwi.t_auction a on a.company_id = c.company_id) d on upper(trim(d.credit_code)) = upper(trim(del.creditcode)) ");
            delete_data.createOrReplaceTempView("tmp_t_auction2");
            spark.sql("insert into table a_dc_ep_incr.t_auction select * from tmp_t_auction2");
        }else {
            spark.sql("TRUNCATE TABLE a_dc_ep_incr.t_auction");
            System.out.println("中数本期数据涉及该表的数据为空。。。。。。");
        }

        spark.stop();
    }
}
